package org.cloudplatform.common.feign.zk;

import com.google.gson.Gson;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.cloudplatform.common.feign.model.Instance;
import org.cloudplatform.common.feign.register.InstanceRegister;

import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 类描述：订阅并更新数据
 * 作者：徐卫超 (cc)
 * 时间 2022/5/17 18:42
 */

@Slf4j
public class ServiceCuratorCacheListener implements CuratorCacheListener, Runnable {

    private Gson gson = new Gson();

    private AtomicBoolean ctl = new AtomicBoolean(false);

    private final InstanceRegister register;
    // 服务注册的根路径
    private final String rootPath;

    private final CuratorFramework curatorFramework;


    public ServiceCuratorCacheListener(InstanceRegister register, String rootPath, CuratorFramework curatorFramework) {
        this.register = register;
        this.rootPath = rootPath;
        this.curatorFramework = curatorFramework;
    }

    /**
     * zk的watch机制是只能监听一次 虽然curator做了高级封装但是还是出现订阅消息丢失的问题，
     * 所有为了保证程序的可靠性，需要定期检查本地缓存和zk的节点数据是否一致。
     */
    @Override
    public void run() {
        Set<String> strings = register.serviceNames();
        log.debug("检查zk服务和本地缓存是否一致 {}", strings);
    }

    @Override
    public void event(Type type, ChildData oldData, ChildData data) {

        if (type.equals(Type.NODE_CREATED)) {  //节点创建
            if (!data.getPath().equals(rootPath)) {
                createService(data);
            }
        } else if (type.equals(Type.NODE_DELETED)) { // 节点修改
            if (!oldData.getPath().equals(rootPath)) {
                delService(oldData);
            }
        } else {
            log.warn("未处理的事件数据：{}", data.toString());
        }
        if (!ctl.get()) {
            ctl.set(true);
        }
    }

    /**
     * 删除服务时被调用
     *
     * @param data 删除的节点
     */
    private void delService(ChildData data) {
        String relativePath = relativePath(data.getPath());
        String[] array = serviceInfo(relativePath);
        if (array.length == 2) { //删除同一类型服务中的一个节点
            Instance instance = convert(data);
            log.debug("实例下线 服务ID: {} 实例ID：{}", instance.getServiceId(), instance.getInstanceId());
            register.remove(instance);
        } else if (array.length == 1) { // 删除同一类型服务中的所有节点
            register.removeServiceName(array[0]);
            log.debug("服务下线：{}", relativePath);
        }
    }

    /**
     * 创建服务被调用
     *
     * @param data 创建的节点信息
     */
    private void createService(ChildData data) {
        String relativePath = relativePath(data.getPath());
        String[] array = serviceInfo(relativePath);
        if (array.length == 2) { //服务注册
            Instance instance = convert(data);
            log.debug("实例注册 服务ID: {} 实例ID：{}", instance.getInstanceId(), instance.getInstanceId());
            register.register(instance);
        } else if (array.length == 1) {
            log.debug("服务注册：{}", relativePath);
        }
    }


    /**
     * 获取注册路径相对路径
     *
     * @param absolutePath 绝对路径
     * @return 返回注册的相对路径
     */
    public String relativePath(String absolutePath) {
        return absolutePath.replaceFirst(rootPath, "");
    }

    /**
     * 获取服务命
     *
     * @param relativePath 服务的相对注册路径
     * @return 返回服务数组  {"serviceName","serviceId"}
     */
    public String[] serviceInfo(String relativePath) {
        String serviceInfoPath = relativePath.startsWith("/") ? relativePath.substring(1) : relativePath;
        return (serviceInfoPath).split("/");
    }


    /**
     * 把zk中的数据转成标准的数据实例
     *
     * @param data zk节点数据
     * @return 标准的实例对象
     */
    private Instance convert(ChildData data) {
        if (data == null) {
            return null;
        }
        ZKNodeData zkNodeData = gson.fromJson(new String(data.getData()), ZKNodeData.class);
        Instance instance = new Instance();
        instance.setServiceId(zkNodeData.getName());
        instance.setInstanceId(zkNodeData.getId());
        instance.setHost(zkNodeData.getAddress());
        instance.setSecure(zkNodeData.getSslPort() != null);
        instance.setPort(instance.isSecure() ? zkNodeData.getSslPort() : zkNodeData.getPort());
        return instance;
    }

    public void isDone() {
        while (!ctl.get()) {
            try {
                TimeUnit.SECONDS.sleep(1l);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * zk 中存储的数据结构
     */
    @Data
    public static class ZKNodeData {
        private String id;
        private String name;
        private String address;
        private Integer port;
        private Integer sslPort;
        private Long registrationTimeUTC;
    }
}
